package com.ds.lens.flink.monitor.forecast.function;

import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.dto.ForecastActualData;
import com.ds.lens.data.common.document.dto.ForecastActualDataPoint;
import com.ds.lens.flink.monitor.common.DataPointComparator;
import com.ds.lens.flink.monitor.common.MonitorDefinitionFactory;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.Collections;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-06
 */
public class ForecastActualDataFunction extends ProcessFunction<MetricBundle, ForecastActualData> {
    private static final long serialVersionUID = 283529736274596517L;

    @Override
    public void processElement(MetricBundle metricBundle, Context context, Collector<ForecastActualData> collector) throws Exception {
        if (metricBundle == null || CollectionUtils.isEmpty(metricBundle.getMetrics())) {
            return;
        }
        MonitorDefinition definition = MonitorDefinitionFactory.get().getMonitorDefinitionByState(metricBundle.getState());
        metricBundle.getMetrics().forEach((key, metric) -> {
            if (!CollectionUtils.isEmpty(metric.getValues())) {
                ForecastActualData input = new ForecastActualData();
                input.setSubTopic(key);
                metric.getValues().sort(new DataPointComparator());
                ForecastActualDataPoint actualDataPoint = new ForecastActualDataPoint();
                DataPoint dataPoint = metric.getValues().get(metric.getValues().size() - 1);
                if (dataPoint.getTimeStamp() == (metricBundle.getEndTime() - definition.getIntervalTime() * 60000)) {
                    actualDataPoint.setTs(dataPoint.getTimeStamp());
                    actualDataPoint.setValue(dataPoint.getValue());
                    input.setData(Collections.singletonList(actualDataPoint));
                    collector.collect(input);
                }
            }
        });
    }
}
